package com.blue.http.server;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * Websocket路由，绑定帐号和链路的关系
 * 
 * @author zhengj
 * @since 1.0 2016年12月15日
 */
public class WebsocketRouter
{
	private static Logger logger = LoggerFactory.getLogger(WebsocketRouter.class);

	private ChannelGroup channelGroup;

	private Map<String, Set<ChannelId>> broadcastMap;
	private Lock broadcastReadLock;
	private Lock broadcastWriteLock;

	private Map<String, Set<Integer>> pointUserMap;
	private Map<Integer, Set<ChannelId>> pointMap;
	private Lock pointReadLock;
	private Lock pointWriteLock;
	
	public WebsocketRouter()
	{
		channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

		broadcastMap = new HashMap<>();
		ReadWriteLock broadcastLock = new ReentrantReadWriteLock();
		broadcastReadLock = broadcastLock.readLock();
		broadcastWriteLock=  broadcastLock.writeLock();

		pointUserMap = new HashMap<>();
		pointMap = new HashMap<>();
		ReadWriteLock pointLock = new ReentrantReadWriteLock();
		pointReadLock = pointLock.readLock();
		pointWriteLock = pointLock.writeLock();
	}

	/**
	 * 添加广播通道
	 * @param name 广播通道名称
	 * @param channel 通道
	 */
	public void addBroadcast(String name, Channel channel)
	{
		Assert.hasText(name, "名称不能为空");
		Assert.notNull(channel, "广播通道不能为空");

		broadcastWriteLock.lock();
		try
		{
			channelGroup.add(channel);
			Set<ChannelId> idSet = broadcastMap.computeIfAbsent(name, (k) -> new HashSet<>());
			idSet.add(channel.id());
			logger.debug("向 {} 广播通道绑定：{}", name, channel.remoteAddress());
		}
		finally
		{
			broadcastWriteLock.unlock();
		}
	}

	/**
	 * 向广播通道发送消息
	 * @param name 广播通道名称
	 * @param object 消息对象
	 */
	public void sendBroacast(String name, Object object)
	{
		Assert.hasText(name, "名称不能为空");
		Assert.notNull(object, "消息对象不能为空");

		broadcastReadLock.lock();
		try
		{
			Set<ChannelId> idSet = broadcastMap.get(name);
			if (idSet == null || idSet.isEmpty())
			{
				logger.info("{} 绑定广播通道为空", name);
				return;
			}

			int i = 0;
			for (ChannelId id : idSet)
			{
				Channel channel = channelGroup.find(id);
				if (channel == null)
					continue;

				channel.writeAndFlush(object);
				logger.debug("向 {} 绑定的广播通道发送消息：{}", name, channel.remoteAddress());
				i++;
			}
			logger.info("向 {} 绑定的广播通道发送消息，个数：{}", name, i);
		}
		finally
		{
			broadcastReadLock.unlock();
		}
	}

	/**
	 * 获取广播通道列表
	 * @param name 广播通道名称
	 * @return 广播通道列表
	 */
	public List<Channel> listBroadcast(String name)
	{
		Assert.hasText(name, "名称不能为空");
		List<Channel> chList = new ArrayList<>();
		broadcastReadLock.lock();
		try
		{
			Set<ChannelId> idSet = broadcastMap.get(name);
			for (ChannelId id : idSet)
			{
				Channel channel = channelGroup.find(id);
				if (channel != null)
				{
					chList.add(channel);
				}
			}
		}
		finally
		{
			broadcastReadLock.unlock();
		}
		return chList;
	}

	/**
	 * 移除通道，包括广播通道和点对点通道
	 * @param channel 通道
	 */
	public void remove(Channel channel)
	{
		Assert.notNull(channel, "通道不能为空");

		Set<String> nameSet = new HashSet<>();
		broadcastWriteLock.lock();
		try
		{
			for (Map.Entry<String, Set<ChannelId>> entry : broadcastMap.entrySet())
			{
				if (entry.getValue() == null || entry.getValue().isEmpty())
					continue;

				entry.getValue().remove(channel.id());
				if (entry.getValue().isEmpty())
				{
					nameSet.add(entry.getKey());
				}
			}
			nameSet.forEach(e -> broadcastMap.remove(e));
		}
		finally
		{
			broadcastWriteLock.unlock();
		}

		nameSet.clear();
		Set<Integer> userSet = new HashSet<>();
		pointWriteLock.lock();
		try
		{
			for (Map.Entry<Integer, Set<ChannelId>> entry : pointMap.entrySet())
			{
				if (entry.getValue() == null || entry.getValue().isEmpty())
					continue;

				entry.getValue().remove(channel.id());
				if (entry.getValue().isEmpty())
				{
					userSet.add(entry.getKey());
				}
			}
			for (Integer user : userSet)
			{
				pointMap.remove(user);

				for (Map.Entry<String, Set<Integer>> entry : pointUserMap.entrySet())
				{
					if (entry.getValue() == null || entry.getValue().isEmpty())
						continue;

					entry.getValue().remove(user);
					if (entry.getValue().isEmpty())
					{
						nameSet.add(entry.getKey());
					}
				}
			}
			nameSet.forEach(e -> pointUserMap.remove(e));
		}
		finally
		{
			pointWriteLock.unlock();
		}
	}

	/**
	 * 添加点对点通道
	 * @param name 点对点通道名称
	 * @param userId 用户ID
	 * @param channel 通道
	 */
	public void addPoint(String name, Integer userId, Channel channel)
	{
		Assert.hasText(name, "名称不能为空");
		Assert.notNull(userId, "用户ID不能为空");
		Assert.notNull(channel, "点对点通道不能为空");

		pointWriteLock.lock();
		try
		{
			channelGroup.add(channel);
			Set<Integer> userSet = pointUserMap.computeIfAbsent(name, (k) -> new HashSet<>());
			userSet.add(userId);

			Set<ChannelId> idSet = pointMap.computeIfAbsent(userId, (k) -> new HashSet<>());
			idSet.add(channel.id());
			logger.debug("向 {} 广播通道绑定用户 {}：{}", name, userId, channel.remoteAddress());
		}
		finally
		{
			pointWriteLock.unlock();
		}
	}

	/**
	 * 向点对点广播发送消息
	 * @param name 点对点通道名称
	 * @param userId 用户ID
	 * @param object 消息对象
	 */
	public void sendPoint(String name, Integer userId, Object object)
	{
		Assert.hasText(name, "名称不能为空");
		Assert.notNull(userId, "用户ID不能为空");
		Assert.notNull(object, "消息对象不能为空");

		pointReadLock.lock();
		try
		{
			Set<Integer> userList = pointUserMap.get(name);
			if (userList == null || userList.isEmpty())
			{
				logger.warn("点对点通道 {} 不存在与用户ID {} 的绑定", name, userId);
				return;
			}

			int i = 0;
			for (Integer user : userList)
			{
				Set<ChannelId> idSet = pointMap.get(user);
				if (idSet == null || idSet.isEmpty())
					continue;

				for (ChannelId id : idSet)
				{
					Channel channel = channelGroup.find(id);
					if (channel == null)
						continue;

					channel.writeAndFlush(object);
					logger.debug("向 {} 绑定的点对点通道发送消息：{}", name, channel.remoteAddress());
					i++;
				}
			}
			logger.info("向 {} 绑定的点对点通道的用户ID {} 发送消息，个数：{}", name, userId, i);
		}
		finally
		{
			pointReadLock.unlock();
		}
	}

	/**
	 * 获取点对点广播通道列表
	 * @param name 点对点通道名称
	 * @param userId 用户ID
	 * @return 点对点通道列表
	 */
	public List<Channel> listPoint(String name, Integer userId)
	{
		Assert.hasText(name, "名称不能为空");
		Assert.notNull(userId, "用户ID不能为空");

		List<Channel> channelList = new ArrayList<>();
		pointReadLock.lock();
		try
		{
			Set<Integer> userList = pointUserMap.get(name);
			if (userList == null || userList.isEmpty())
				return channelList;

			for (Integer user : userList)
			{
				Set<ChannelId> idSet = pointMap.get(user);
				if (idSet == null || idSet.isEmpty())
					continue;

				for (ChannelId id : idSet)
				{
					Channel channel = channelGroup.find(id);
					if (channel == null)
						continue;

					channelList.add(channel);
				}
			}
		}
		finally
		{
			pointReadLock.unlock();
		}
		return channelList;
	}
	
}
